package com.atguigu.actual.edu0417.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.unils.DimInfoUtil;
import com.atguigu.actual.edu0417.unils.HbaseUtil;
import com.atguigu.actual.edu0417.unils.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.util.Objects;

/**
 * @author: techyang
 * @since: 2023-10-21 23:57
 * @description:
 **/
public abstract class DimMapFunction<T> extends RichMapFunction<T,T> implements DimJoinFunction<T> {

    // HBase连接
    private Connection conn;
    // Redis客户端
    private Jedis jedis;
    // 表名
    private String tableName;

    public DimMapFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getHbaseConnection();
        jedis = RedisUtil.getRedisClient();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(conn);
        RedisUtil.closeRedisClient(jedis);
    }

    @Override
    public T map(T bean) throws Exception {
        String key = getKey(bean);
        JSONObject dimCourseJsonObj = DimInfoUtil.getDimInfo(jedis, conn, EduConfig.HBASE_NAMESPACE, tableName, key);
        if (Objects.nonNull(dimCourseJsonObj)) {
            join(bean,dimCourseJsonObj);
            return bean;
        }
        return null;
    }

}
